package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/8
 * @Desc: 学大数据，上多易教育
 * <p>
 * 模型2规则示例：
 * 静态画像条件：
 * age>10,gender=male
 * 动态画像统计条件：
 * X>=3次
 * 触发条件：
 * Y行为
 * <p>
 * 参数模板：
{
"rule_id":"rule-m02-004",
"static_profile_conditions":[
{
"tag_name":"age",
"compare_type":">",
"compare_value":10
},
{
"tag_name":"gender",
"compare_type":"=",
"compare_value":"male"
}
],
"realtime_statistic_conditions":{
"event_id":"X",
"min_count":3
},
"trigger_event_id":"Y"
}
 **/
public class RuleModelCalculator_02 implements RuleCalculator {

    private String ruleId;
    Roaring64Bitmap preSelectBitmap;
    ValueState<Integer> state;

    JSONObject paramJsonObject;
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext, RuleMeta ruleMeta) throws IOException {


        String ruleParamJson = ruleMeta.getRule_param_json();

        this.paramJsonObject = JSON.parseObject(ruleParamJson);
        this.ruleId = paramJsonObject.getString("rule_id");

        this.message = new JSONObject();
        message.put("rule_id",ruleId);

        this.preSelectBitmap = ruleMeta.getPreSelectBitmap();

        state = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-cnt", Integer.class));
    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        // 判断该行为人，是否属于满足静态画像条件的人[从bitmap中判断即可]
        if(!preSelectBitmap.contains(userAction.getUser_id())) return;


        String eventId = userAction.getEvent_id();

        JSONObject realtimeStatisticConditions = paramJsonObject.getJSONObject("realtime_statistic_conditions");
        String targetEventId = realtimeStatisticConditions.getString("event_id");
        int minCount = realtimeStatisticConditions.getIntValue("min_count");

        // 动态画像条件统计
        if(eventId.equals(targetEventId)){
            state.update( state.value()==null? 1: state.value()+1 );
        }



        // 触发判断
        String triggerEventId = paramJsonObject.getString("trigger_event_id");
        if(eventId.equals(triggerEventId)){
            if(state.value() != null && state.value()>= minCount){
                message.put("user_id",userAction.getUser_id());
                message.put("hit_time",userAction.getAction_time());
                out.collect(message.toJSONString());
            }
        }
    }

    @Override
    public String ruleId() {
        return ruleId;
    }

    @Override
    public boolean isNeedProcessPastData() {
        return false;
    }


}
